/**
 * Copyright (C) 2010, 2011 Neofonie GmbH
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package eu.dicodeproject.twitterstream;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

import twitter4j.Status;
import eu.dicodeproject.twitterstream.sink.TweetSink;
import eu.dicodeproject.twitterstream.source.TweetSource;

/**
 * Takes incoming objects from tweet source and puts them to the tweet sink.
 * Runs multiple threads if configured that way.
 */
public class StreamingBroker {
  /** Source to take tweets from. */
  private final Iterator<Status> sourceIt;
  /** Sink to store tweets to. */
  private final TweetSink sink;
  /*** Thread pool for retrieving and storing tweets to HBase. */
  private final ExecutorService executor;
  /** Indicates if routing started already. */
  private boolean routing = false;
  /** List of pointers to broker results. */
  private List<Future<Boolean>> results = new ArrayList<Future<Boolean>>();
  private TweetSource source;
  private int streamBrokerPoolSize;

  /**
   * Init the broker. Thread pool sizes can be configured by properties file.
   * See also TwitterHarvesterCFG implementation.
   * 
   * @param sourceIt
   *          the tweet sourceIt to take tweets from.
   * @param sink
   *          the sink to store tweets to.
   * */
  public StreamingBroker(final TweetSource source, final TweetSink sink, int streamBrokerPoolSize) {
    this.sink = sink;
    this.source = source;
    this.sourceIt = source.iterator();
    this.executor = Executors.newFixedThreadPool(streamBrokerPoolSize);
    this.streamBrokerPoolSize = streamBrokerPoolSize;
  }

  public StreamingBroker(final TweetSource source, final TweetSink sink) {
    this(source, sink, 10);
  }

  /** Initiate tweet routing. If called twice, no state is changed. */
  public synchronized void start() {
    this.source.triggerHarvesting();
    if (!this.routing) {
      int threadNum = this.streamBrokerPoolSize;
      for (int i = 0; i < threadNum; i++) {
        Future<Boolean> result = this.executor.submit(new Plane(sourceIt, sink));
        this.results.add(result);
      }
      this.routing = true;
    }
  }

  /** Stop storing tweets in HBase. If called twice no state is changed. */
  public synchronized void stop() {
    if (this.routing) {
      for (Future<Boolean> future : results) {
        future.cancel(true);
      }
    }
  }

  /**
   * Returns true if there are still threads carrying tweets from the api to the
   * sink. False otherwise.
   */
  public boolean busy() {
    for (Future<Boolean> result : results) {
      if (!result.isDone() && !result.isCancelled()) {
        return true;
      }
    }
    return false;
  }

  private static final class Plane implements Callable<Boolean> {
    /** Source iterator. */
    private final Iterator<Status> source;
    /** Sink for tweets. */
    private final TweetSink sink;

    public Plane(final Iterator<Status> source, final TweetSink sink) {
      this.source = source;
      this.sink = sink;
    }

    /** {@inheritDoc} */
    @Override
    public Boolean call() throws InterruptedException, IOException {
      while (source.hasNext()) {
        Status status = source.next();
        if (status != null) {
          this.sink.store(status);
        }
        Thread.sleep(200);
      }
      return Boolean.TRUE;
    }
  }
}
